package com.lm.paymentadmin.service;

import com.alibaba.fastjson.JSON;
import com.lm.paymentadmin.annotation.FieldType;
import com.lm.paymentadmin.annotation.Index;
import com.lm.paymentadmin.common.bean.PageEntity;
import com.lm.paymentadmin.domain.EsEntity;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.DocWriteRequest;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.*;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.client.indices.GetIndexResponse;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.IOException;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@Service
@Slf4j
public class EsService {

    @Resource
    private RestHighLevelClient highLevelClient;

    public PageEntity search(PageEntity page, SearchSourceBuilder searchSourceBuilder,Class clazz){
        try {
            if (clazz.isAnnotationPresent(Index.class)) {
                Index index = (Index) clazz.getAnnotation(Index.class);
                SearchRequest searchRequest = new SearchRequest(index.value());
                searchSourceBuilder.from((page.getPage()-1)*page.getPageSize());
                searchSourceBuilder.size(page.getPageSize());
                searchRequest.source(searchSourceBuilder);
                SearchResponse search = highLevelClient.search(searchRequest, RequestOptions.DEFAULT);
                List<Object> list = new ArrayList<>();
                if (search.getHits().getHits().length!=0){
                    for (SearchHit documentFields : search.getHits().getHits()) {
                        list.add(JSON.parseObject(documentFields.getSourceAsString(),clazz));
                    }
                }
                page.setTotalCount(search.getHits().getTotalHits().value);
                long totalPages = search.getHits().getTotalHits().value / page.getPageSize();
                if (search.getHits().getTotalHits().value % page.getPageSize() != 0){
                    totalPages ++;
                }
                page.setTotalPage(totalPages);
                page.setDataList(list);
            }
        }catch (IOException e){
            e.printStackTrace();
        }
        return page;
    }

    /**
     * 创建索引
     * @param clazz
     * @throws IOException
     */
    public void createContentIk(Class clazz) throws IOException {
        if (clazz.isAnnotationPresent(Index.class)) {
            Index index = (Index) clazz.getAnnotation(Index.class);
            //判断是否存在该索引
            if(isExistsIndex(index.value())){
                log.info("索引:{}已经存在",index.value());
//                //删除
//                deleteIndex(index.value());
            }else {
                //这里的都是es的对象 indexName就是你的索引名
                CreateIndexRequest request = new CreateIndexRequest(index.value());
                Field[] fields = clazz.getDeclaredFields();
                Map<String,Object> properties = new HashMap<>();
                for (Field field : fields) {
                    if(field.isAnnotationPresent(FieldType.class)){
                        properties.put(field.getName(),getField(field.getAnnotation(FieldType.class).value()));
                    }
                }
                Map<String,Object> mapping = new HashMap<>();
                mapping.put("properties", properties);
                request.mapping(mapping);
                // 执行创建请求
                CreateIndexResponse createIndexResponse = highLevelClient.indices().create(request, RequestOptions.DEFAULT);
                //判断成功与否
                boolean acknowledged = createIndexResponse.isAcknowledged();
                log.info("acknowledged:{}", acknowledged);
                boolean shardsAcknowledged = createIndexResponse.isShardsAcknowledged();
                log.info("shardsAcknowledged:{}", shardsAcknowledged);
            }

        }

    }

    /**
     * 查询索引信息
     * @param clazz
     * @return
     */
    public String findIndex(Class clazz){
        if (clazz.isAnnotationPresent(Index.class)) {
            Index index = (Index) clazz.getAnnotation(Index.class);
            GetIndexRequest request = new GetIndexRequest(index.value());
            HashMap<String, Object> result = new HashMap<>();
            try {
                GetIndexResponse getIndexResponse = highLevelClient.indices().get(request, RequestOptions.DEFAULT);
                Map<String, Settings> settings = getIndexResponse.getSettings();
                log.info("index settings is :" + settings.toString());
                return settings.toString();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return "";
    }

    /**
     * 判断一个索引是否存在
     * @param indexName
     * @return
     */
    private  boolean isExistsIndex(String indexName){
        GetIndexRequest request = new GetIndexRequest(indexName);
        try {
            boolean exists = highLevelClient.indices().exists(request, RequestOptions.DEFAULT);
            return exists;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 删除整个索引
     * @param indexName
     */
    private void deleteIndex(String indexName){
        try {
            DeleteIndexRequest info = new DeleteIndexRequest(indexName);
            AcknowledgedResponse delete = highLevelClient.indices().delete(info, RequestOptions.DEFAULT);
            log.info("delete index result is : " + delete.isAcknowledged());
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 单条插入
     * @param o
     * @throws IOException
     */
    public void singleAdd(Object o) {
        try {
            if (o.getClass().isAnnotationPresent(Index.class)) {
                Index index =  o.getClass().getAnnotation(Index.class);
                IndexRequest indexRequest = new IndexRequest(index.value());
                indexRequest.source(JSON.toJSONString(o), XContentType.JSON);
                indexRequest.timeout(TimeValue.timeValueSeconds(1));
                indexRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);
                indexRequest.create(true);
                EsEntity esEntity = (EsEntity)o;
                indexRequest.id(esEntity.getId() + "");
                highLevelClient.index(indexRequest, RequestOptions.DEFAULT);
            }
        }catch (IOException e){
            e.printStackTrace();
        }
    }

    /**
     * 单条异步插入
     * @param o
     * @throws IOException
     */
    public void singleAsyncAdd(Object o) {
        if (o.getClass().isAnnotationPresent(Index.class)) {
            Index index =  o.getClass().getAnnotation(Index.class);
            IndexRequest indexRequest = new IndexRequest(index.value());
            indexRequest.source(JSON.toJSONString(o), XContentType.JSON);
            indexRequest.timeout(TimeValue.timeValueSeconds(1));
            indexRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);
            //数据为存储而不是更新
            indexRequest.create(false);
            EsEntity esEntity = (EsEntity)o;
            indexRequest.id(esEntity.getId() + "");
            highLevelClient.indexAsync(indexRequest, RequestOptions.DEFAULT, new ActionListener<IndexResponse>() {
                @Override
                public void onResponse(IndexResponse indexResponse) {
                    ReplicationResponse.ShardInfo shardInfo = indexResponse.getShardInfo();
                    if (shardInfo.getFailed() > 0) {
                        for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                            log.error("将id为：{}的数据存入ES时存在失败的分片，原因为：{}", indexRequest.id(), failure.getCause());
                        }
                    }
                }
                @Override
                public void onFailure(Exception e) {
                    log.error("{}:存储es时异常，数据信息为", indexRequest.id(), e);
                }
            });
        }
    }
    /**
     * 批量插入
     * @param list
     */
    public void batchAdd(List<Object> list){
        List<IndexRequest> indexRequests = new ArrayList<>();
        list.forEach(e -> {
            if (e.getClass().isAnnotationPresent(Index.class)) {
                Index index =  e.getClass().getAnnotation(Index.class);
                IndexRequest request = new IndexRequest(index.value());
                EsEntity esEntity = (EsEntity)e;
                //填充id
                request.id(esEntity.getId() + "");
                //先不修改id
                request.source(JSON.toJSONString(e), XContentType.JSON);
                request.opType(DocWriteRequest.OpType.CREATE);
                indexRequests.add(request);
            }
        });
        indexRequests.forEach(createBulkProcessor()::add);
    }

    /**
     * 单条更新
     * @param e
     */
    public void singleUpdate(Object e) {
        try {
            if (e.getClass().isAnnotationPresent(Index.class)) {
                Index index =  e.getClass().getAnnotation(Index.class);
                UpdateRequest request = new UpdateRequest();
                EsEntity esEntity = (EsEntity)e;
                request.index(index.value());
                //填充id
                request.id(esEntity.getId() + "");
                //更新的数据
                Field[] fields = e.getClass().getDeclaredFields();
                Map<String,Object> map=new HashMap<>();
                for (Field field : fields) {
                    field.setAccessible(true);
                    if(field.isAnnotationPresent(FieldType.class)){
                        try {
                            map.put(field.getName(),field.get(e));
                        } catch (IllegalAccessException ex) {
                            ex.printStackTrace();
                        }
                    }
                }
                request.doc(map);
                highLevelClient.update(request, RequestOptions.DEFAULT);
            }
        }catch (IOException ex){
            ex.printStackTrace();
        }
    }

    /**
     * 批量更新
     * @param list
     */
    public void batchUpdate(List<Object> list){
        List<UpdateRequest> updateRequests = new ArrayList<>();
        list.forEach(e -> {
            if (e.getClass().isAnnotationPresent(Index.class)) {
                Index index =  e.getClass().getAnnotation(Index.class);
                UpdateRequest request = new UpdateRequest();
                EsEntity esEntity = (EsEntity)e;
                request.index(index.value());
                //填充id
                request.id(esEntity.getId() + "");
                //更新的数据
                Field[] fields = e.getClass().getDeclaredFields();
                Map<String,Object> map=new HashMap<>();
                for (Field field : fields) {
                    field.setAccessible(true);
                    if(field.isAnnotationPresent(FieldType.class)){
                        try {
                            map.put(field.getName(),field.get(e));
                        } catch (IllegalAccessException ex) {
                            ex.printStackTrace();
                        }
                    }
                }
                request.doc(map);
                updateRequests.add(request);
            }
        });
        updateRequests.forEach(createBulkProcessor()::add);
    }

    /**
     *
     * @param type
     * @return
     */
    private static Map<String, Object> getField(String type) {
        Map<String, Object> result = new HashMap<>();
        result.put("type", type);
        if ("text".equals(type)) {
            result.put("analyzer", "whitespace");
            result.put("search_analyzer", "whitespace");
        }
        return result;
    }


    /**
     * 重写BulkProcessor
     * @return
     */
    public BulkProcessor createBulkProcessor() {

        BulkProcessor.Listener listener = new BulkProcessor.Listener() {
            @Override
            public void beforeBulk(long executionId, BulkRequest request) {
                log.info("1. 【beforeBulk】批次[{}] 携带 {} 请求数量", executionId, request.numberOfActions());
            }

            @Override
            public void afterBulk(long executionId, BulkRequest request,
                                  BulkResponse response) {
                if (!response.hasFailures()) {
                    log.info("2. 【afterBulk-成功】批量 [{}] 完成在 {} ms", executionId, response.getTook().getMillis());
                } else {
                    BulkItemResponse[] items = response.getItems();
                    for (BulkItemResponse item : items) {
                        if (item.isFailed()) {
                            log.info("2. 【afterBulk-失败】批量 [{}] 出现异常的原因 : {}", executionId, item.getFailureMessage());
                            break;
                        }
                    }
                }
            }

            @Override
            public void afterBulk(long executionId, BulkRequest request,
                                  Throwable failure) {

                List<DocWriteRequest<?>> requests = request.requests();
                List<String> esIds = requests.stream().map(DocWriteRequest::id).collect(Collectors.toList());
                log.error("3. 【afterBulk-failure失败】es执行bluk失败,失败的esId为：{}", esIds, failure);
            }
        };

        BulkProcessor.Builder builder = BulkProcessor.builder(((bulkRequest, bulkResponseActionListener) -> {
            highLevelClient.bulkAsync(bulkRequest, RequestOptions.DEFAULT, bulkResponseActionListener);
        }), listener);
        //到达10000条时刷新
        builder.setBulkActions(10000);
        //内存到达8M时刷新
        builder.setBulkSize(new ByteSizeValue(8L, ByteSizeUnit.MB));
        //设置的刷新间隔10s
        builder.setFlushInterval(TimeValue.timeValueSeconds(10));
        //设置允许执行的并发请求数。
        builder.setConcurrentRequests(8);
        //设置重试策略
        builder.setBackoffPolicy(BackoffPolicy.constantBackoff(TimeValue.timeValueSeconds(1), 3));
        return builder.build();
    }


}
